package com.atguigu.tingshu.album.service.impl;

import com.atguigu.tingshu.album.mapper.AlbumAttributeValueMapper;
import com.atguigu.tingshu.album.mapper.AlbumInfoMapper;
import com.atguigu.tingshu.album.mapper.AlbumStatMapper;
import com.atguigu.tingshu.album.service.AlbumInfoService;
import com.atguigu.tingshu.album.util.RedisLock;
import com.atguigu.tingshu.album.util.RedisLockClient;
import com.atguigu.tingshu.common.cache.TingshuCache;
import com.atguigu.tingshu.common.cache.TingshuCacheAspect;
import com.atguigu.tingshu.common.constant.KafkaConstant;
import com.atguigu.tingshu.common.constant.RedisConstant;
import com.atguigu.tingshu.common.constant.SystemConstant;
import com.atguigu.tingshu.common.service.KafkaService;
import com.atguigu.tingshu.common.util.AuthContextHolder;
import com.atguigu.tingshu.model.album.AlbumAttributeValue;
import com.atguigu.tingshu.model.album.AlbumInfo;
import com.atguigu.tingshu.model.album.AlbumStat;
import com.atguigu.tingshu.query.album.AlbumInfoQuery;
import com.atguigu.tingshu.vo.album.AlbumAttributeValueVo;
import com.atguigu.tingshu.vo.album.AlbumInfoVo;
import com.atguigu.tingshu.vo.album.AlbumListVo;
import com.atguigu.tingshu.vo.album.AlbumStatVo;
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.baomidou.mybatisplus.core.conditions.update.LambdaUpdateWrapper;
import com.baomidou.mybatisplus.core.metadata.IPage;
import com.baomidou.mybatisplus.core.toolkit.CollectionUtils;
import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import lombok.extern.slf4j.Slf4j;
import org.redisson.api.RBloomFilter;
import org.redisson.api.RLock;
import org.redisson.api.RedissonClient;
import org.springframework.beans.BeanUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.autoconfigure.cache.CacheProperties;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.core.script.DefaultRedisScript;
import org.springframework.stereotype.Service;

import java.util.Arrays;
import java.util.List;
import java.util.Random;
import java.util.UUID;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;

@Slf4j
@Service
@SuppressWarnings({"unchecked", "rawtypes"})
public class AlbumInfoServiceImpl extends ServiceImpl<AlbumInfoMapper, AlbumInfo> implements AlbumInfoService {

    @Autowired
    private AlbumInfoMapper albumInfoMapper;

    @Autowired
    private AlbumAttributeValueMapper albumAttributeValueMapper;

    @Autowired
    private AlbumStatMapper albumStatMapper;

    @Autowired
    private KafkaService kafkaService;

    @Autowired
    private RedisTemplate redisTemplate;

    @Autowired
    private RedisLockClient redisLockClient;

    @Autowired
    private RedissonClient redissonClient;

    @Override
    public void saveAlbumInfo(AlbumInfo albumInfo) {
        //保存专辑表数据
        Long userId = AuthContextHolder.getUserId();
        userId = userId == null ? 1L : userId;
        albumInfo.setUserId(userId);
        albumInfo.setIncludeTrackCount(0);
        albumInfo.setIsFinished(0);
        albumInfo.setQualityScore("0");
        if (!SystemConstant.ALBUM_PAY_TYPE_FREE.equals(albumInfo.getPayType())) {
            albumInfo.setTracksForFree(5);
            albumInfo.setSecondsForFree(30);
        }
        albumInfo.setStatus("0301");
        this.save(albumInfo);
        Long albumId = albumInfo.getId();
        //保存标签表数据
        List<AlbumAttributeValue> albumAttributeValueVoList = albumInfo.getAlbumAttributeValueVoList();

        if (!CollectionUtils.isEmpty(albumAttributeValueVoList)) {
            albumAttributeValueVoList.forEach(albumAttributeValue -> {
                albumAttributeValue.setAlbumId(albumId);
                albumAttributeValueMapper.insert(albumAttributeValue);
            });
        }
        //保存统计表数据
        getAlbumStat(albumId, SystemConstant.ALBUM_STAT_PLAY);
        getAlbumStat(albumId, SystemConstant.ALBUM_STAT_SUBSCRIBE);
        getAlbumStat(albumId, SystemConstant.ALBUM_STAT_BROWSE);
        getAlbumStat(albumId, SystemConstant.ALBUM_STAT_COMMENT);

        //同步更新索引库数据
        if ("1".equals(albumInfo.getIsOpen())) {
            kafkaService.sendMsg(KafkaConstant.QUEUE_ALBUM_UPPER, albumId.toString());
        }
        //添加数据时，向布隆过滤器中添加数据
        RBloomFilter<String> bloomFilter = this.redissonClient.getBloomFilter(RedisConstant.ALBUM_BLOOM_FILTER);
        bloomFilter.add(RedisConstant.ALBUM_INFO_PREFIX + albumId);
    }

    private void getAlbumStat(Long albumId, String statType) {
        AlbumStat albumStat = new AlbumStat();
        albumStat.setAlbumId(albumId);
        albumStat.setStatType(statType);
        albumStat.setStatNum(0);
        albumStatMapper.insert(albumStat);
    }


    @Override
    public IPage<AlbumListVo> findUserAlbumPage(Integer page, Integer limit, AlbumInfoQuery albumInfoQuery) {
        Page<AlbumListVo> pages = new Page<>(page, limit);
        Long userId = AuthContextHolder.getUserId();
        userId = userId == null ? 1L : userId;
        albumInfoQuery.setUserId(userId);
        return albumInfoMapper.findUserAlbumPage(pages, albumInfoQuery);
    }


    @Override
    public void removeAlbumInfo(Long albumId) {
        //删除专辑
        this.removeById(albumId);
        //删除标签表
        albumAttributeValueMapper.delete(new LambdaUpdateWrapper<AlbumAttributeValue>().eq(AlbumAttributeValue::getAlbumId, albumId));
        //删除统计表
        albumStatMapper.delete(new LambdaUpdateWrapper<AlbumStat>().eq(AlbumStat::getAlbumId, albumId));
        //同步删除索引库
        kafkaService.sendMsg(KafkaConstant.QUEUE_ALBUM_LOWER, albumId.toString());
    }

    @Override
    public List<AlbumInfo> findUserAllAlbumList() {
        Long userId = AuthContextHolder.getUserId();
        userId = userId == null ? 1L : userId;
        List<AlbumInfo> list = this.list(new LambdaQueryWrapper<>(AlbumInfo.class).eq(AlbumInfo::getUserId, userId)
                .select(AlbumInfo::getId, AlbumInfo::getAlbumTitle).orderByDesc(AlbumInfo::getId));
        return list;
    }

    @TingshuCache(prefix = RedisConstant.ALBUM_INFO_PREFIX,
            lockPrefix = RedisConstant.ALBUM_LOCK_SUFFIX,
            timeout = RedisConstant.CACHE_TIMEOUT,
            random = RedisConstant.ALBUM_TIMEOUT,
            bloomKey = RedisConstant.ALBUM_BLOOM_FILTER)
    @Override
    public AlbumInfo getAlbumInfo(Long albumId) {
        //查询专辑信息
        AlbumInfo albumInfo = this.getById(albumId);
        //查询标签信息
        List<AlbumAttributeValue> albumAttributeValues = albumAttributeValueMapper.selectList(new LambdaQueryWrapper<AlbumAttributeValue>()
                .eq(AlbumAttributeValue::getAlbumId, albumId));
        albumInfo.setAlbumAttributeValueVoList(albumAttributeValues);
        return albumInfo;
    }

    //    @Override
   /* public AlbumInfo getAlbumInfo2(Long albumId) {
        //查询缓存如果命中直接返回，未命中查询数据库
        AlbumInfo albumInfo = (AlbumInfo) this.redisTemplate.opsForValue().get(RedisConstant.ALBUM_INFO_PREFIX + albumId);
        if (albumInfo != null) {
            return albumInfo;
        }
        //加分布式锁 解决缓存击穿问题
        RLock lock = redissonClient.getLock(RedisConstant.ALBUM_LOCK_SUFFIX + albumId);
        lock.lock();
        try {
            //再查缓存 可能第一次查询数据库已经将数据放入了缓存
            albumInfo = (AlbumInfo) this.redisTemplate.opsForValue().get(RedisConstant.ALBUM_INFO_PREFIX + albumId);
            if (albumInfo != null) {
                return albumInfo;
            }
            //查询专辑信息
            albumInfo = this.getById(albumId);
            //查询标签信息
            List<AlbumAttributeValue> albumAttributeValues = albumAttributeValueMapper.selectList(new LambdaQueryWrapper<AlbumAttributeValue>()
                    .eq(AlbumAttributeValue::getAlbumId, albumId));
            albumInfo.setAlbumAttributeValueVoList(albumAttributeValues);
            //设置随机的过期时间 解决缓存雪崩问题
            this.redisTemplate.opsForValue().set(RedisConstant.ALBUM_INFO_PREFIX + albumId, albumInfo, RedisConstant.CACHE_TIMEOUT + new Random().nextInt(5 * 3600), TimeUnit.SECONDS);
            return albumInfo;
        } finally {
            lock.unlock();
        }
    }*/

    @Override
    public void updateAlbumInfo(Long albumId, AlbumInfoVo albumInfoVo) {
        //修改专辑
        AlbumInfo albumInfo = new AlbumInfo();
        BeanUtils.copyProperties(albumInfoVo, albumInfo);
        albumInfo.setId(albumId);
        this.updateById(albumInfo);
        //修改标签
        albumAttributeValueMapper.delete(new LambdaQueryWrapper<AlbumAttributeValue>().eq(AlbumAttributeValue::getAlbumId, albumId));
        List<AlbumAttributeValueVo> list = albumInfoVo.getAlbumAttributeValueVoList();
        if (!CollectionUtils.isEmpty(list)) {
            list.forEach(albumAttributeValueVo -> {
                AlbumAttributeValue albumAttributeValue = new AlbumAttributeValue();
                BeanUtils.copyProperties(albumAttributeValueVo, albumAttributeValue);
                albumAttributeValue.setAlbumId(albumId);
                albumAttributeValueMapper.insert(albumAttributeValue);
            });
        }
        //同步更新索引库
        if ("1".equals(albumInfoVo.getIsOpen())) {
            kafkaService.sendMsg(KafkaConstant.QUEUE_ALBUM_UPPER, albumId.toString());
        }
    }


    @Override
    public List<AlbumInfo> findAlbumPage(Integer page, Integer limit) {
        return this.page(new Page<>(page, limit)).getRecords();
    }


    @Override
    public AlbumStatVo findAlbumStat(Long albumId) {
        return albumStatMapper.findAlbumStat(albumId);
    }

    @Override
    public List<AlbumAttributeValue> findAttributes(Long albumId) {
        return this.albumAttributeValueMapper.selectList(new LambdaQueryWrapper<AlbumAttributeValue>().eq(AlbumAttributeValue::getAlbumId, albumId));
    }

//    RedisLock lock;

    @Override
    public void testLock() {
//        lock = new RedisLock(redisTemplate, "lock");
        RedisLock lock = redisLockClient.getLock("lock");
//        RLock lock = redissonClient.getLock("lock");
        try {
            lock.lock();
            Object num = this.redisTemplate.opsForValue().get("num");
            if (num == null) {
                this.redisTemplate.opsForValue().set("num", 1);
                return;
            }
            Integer count = Integer.valueOf(num.toString());
            this.redisTemplate.opsForValue().set("num", ++count);
//            try {
//                TimeUnit.SECONDS.sleep(1000);
//            } catch (InterruptedException e) {
//                throw new RuntimeException(e);
//            }
            testSubLock();
        } finally {
            lock.unlock();
        }
    }

    public void testSubLock() {
        RedisLock lock = redisLockClient.getLock("lock");
//        RLock lock = redissonClient.getLock("lock");
        lock.lock();
        System.out.println("=========================");
        lock.unlock();
    }

    public void testLock2() {
//        Boolean flag = this.redisTemplate.opsForValue().setIfAbsent("lock", "111");
        //设置过期时间 保证原子性
        //存入值唯一
        String uuid = UUID.randomUUID().toString();
        Boolean flag = this.redisTemplate.opsForValue().setIfAbsent("lock", uuid, 3, TimeUnit.SECONDS);
        if (!flag) {
            try {
                Thread.sleep(40);
                this.testLock();
            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            }
        } else {
            Object num = this.redisTemplate.opsForValue().get("num");
            if (num == null) {
                this.redisTemplate.opsForValue().set("num", 1);
                return;
            }
            Integer count = Integer.valueOf(num.toString());
            this.redisTemplate.opsForValue().set("num", ++count);
            //释放锁 会导致误删 使用lua脚本
//            this.redisTemplate.delete("lock");
            String statement = "if redis.call('get',KEYS[1])==ARGV[1] then return redis.call('del',KEYS[1]) else return 0 end";
            this.redisTemplate.execute(new DefaultRedisScript(statement, Boolean.class), Arrays.asList("lock"), uuid);
        }
    }

    @Override
    public List<Long> getChangedAlbumIds(String startTime, String endTime) {
        List<AlbumStat> albumStats = albumStatMapper.selectList(new LambdaQueryWrapper<AlbumStat>()
                .gt(AlbumStat::getUpdateTime, startTime)
                .le(AlbumStat::getUpdateTime, endTime)
                .select(AlbumStat::getAlbumId));
        if (CollectionUtils.isEmpty(albumStats)) {
            return null;
        }
        return albumStats.stream().distinct().map(albumStat -> albumStat.getAlbumId()).collect(Collectors.toList());
    }

    @Override
    public List<AlbumStatVo> getStatByIds(List<Long> ids) {
        return albumStatMapper.getStatByIds(ids);
    }
}
